package com.example.kafka.service;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.apache.kafka.common.header.internals.RecordHeaders;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;

import java.nio.charset.StandardCharsets;
import java.util.List;


/**
 * | 概念       | 作用                               |
 * | --------- | -------------------------------- |
 * | **Topic** | 消息分类容器（生产者写入、消费者读取）              |
 * | **Group** | 消费者逻辑分组，控制消费并发与 offset 提交        |
 * | **Key**   | 控制消息落入哪个 Partition（分区），从而控制消费顺序性 |
 * 丨
 * Kafka消费者服务
 *
 * @author wkj
 */
@Service
@Slf4j
public class KafkaConsumerService {

    private final KafkaTemplate<String, String> kafkaTemplate;

    public KafkaConsumerService(KafkaTemplate<String, String> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
        this.kafkaTemplate.setTransactionIdPrefix("txn-");
    }

    /**
     * 监听单个消息
     *
     * @param consumeRecord 消费的消息记录
     */
    @KafkaListener(topics = "${spring.kafka.template.default-topic}", groupId = "${spring.kafka.consumer.group-id}")
    public void listenSingleMessage(ConsumerRecord<String, String> consumeRecord) {
        log.info("监听单个消息，主题：{}，key：{}，value: {}",
                consumeRecord.topic(), consumeRecord.key(), consumeRecord.value());
    }

    /**
     * 监听批量消息
     *
     * @param consumeRecords 批量消息记录
     * @param ack            手动确认对象（当配置为手动确认时使用）
     */
    /*
      containerFactory = "kafkaListenerContainerFactory"：使用哪个 Kafka 监听容器工厂（ConcurrentKafkaListenerContainerFactory）来创建消费者监听容器。
      该容器 在 KafkaConsumerConfig 配置中配置了批量消费，因此这里使用 kafkaListenerContainerFactory
     */
    @KafkaListener(topics = "my-batch", containerFactory = "kafkaListenerContainerFactory")
    public void listenBatchMessages(List<ConsumerRecord<String, String>> consumeRecords, Acknowledgment ack) {
        try {
            log.info("批量消息数量: {}", consumeRecords.size());
            for (ConsumerRecord<String, String> consumeRecord : consumeRecords) {
                log.info("监听单个消息，主题：{}，key：{}，value: {}，偏移量：{}，分区：{}",
                        consumeRecord.topic(), consumeRecord.key(), consumeRecord.value(), consumeRecord.offset(), consumeRecord.partition());
                // 处理逻辑 ...
            }
            // 处理成功后手动 ack
            ack.acknowledge();
        } catch (Exception e) {
            log.error("处理失败，提交 offset，失败原因：{}", e.getMessage());
            // 不调用 ack.acknowledge()，Kafka 将在下一次重新投递
        }
    }

    /**
     * 监听特定主题的消息 （类似于 监听单个消息， 只需要自己设置topics，和groupId）
     *
     * @param message 消息内容
     */
    @KafkaListener(topics = "custom-topic", groupId = "custom-group")
    public void listenCustomTopic(String message) {
        log.info("Received message from custom topic: {}", message);
    }

    /**
     * 手动提交偏移量示例
     *
     * @param consumeRecord 消费者记录
     * @param ack           手动确认对象
     */
    @KafkaListener(topics = "reliable-topic", groupId = "reliable-group")
    public void consumeReliably(ConsumerRecord<String, String> consumeRecord, Acknowledgment ack) {
        try {
            log.info("手动提交偏移量示例，主题：{}，key：{}，value: {}",
                    consumeRecord.topic(), consumeRecord.key(), consumeRecord.value());
            // 模拟业务处理
            // ..........
            // 手动提交偏移量
            ack.acknowledge();
            log.info("已提交偏移量");
        } catch (Exception e) {
            // 不提交偏移量，消息会被重新消费
            log.error("不提交偏移量，消息会被重新消费，失败原因：{}", e.getMessage());
        }
    }

    @KafkaListener(topics = "dead-letter-topic", groupId = "dlt-group")
    public void listenDLT(ConsumerRecord<String, String> consumeRecord) {
        log.warn("死信队列接收到失败消息:Key:{},value:{}", consumeRecord.key(), consumeRecord.value());
    }

    private static final int MAX_RETRIES = 3;
    private static final String DEAD_LETTER_TOPIC = "dead-letter-topic";

    /**
     * 消费者重试机制，适用于消费失败的消息，会根据重试次数进行重试。
     * 超过最大重试次数后，消息会被发送到死信队列。
     *
     * @param consumeRecord 消费者记录
     * @param ack           手动确认对象
     */
    @KafkaListener(topics = "retryable-topic", groupId = "retryable-group")
    public void consumeWithRetry(ConsumerRecord<String, String> consumeRecord, Acknowledgment ack) {
        // 从消息头中提取重试次数
        int retryCount = extractRetryCount(consumeRecord.headers());

        try {
            log.info("手动提交偏移量示例，主题：{}，key：{}，value: {}，重试次数：{}",
                    consumeRecord.topic(), consumeRecord.key(), consumeRecord.value(), retryCount);
            // 业务处理
            // .......
            ack.acknowledge();
        } catch (Exception e) {
            retryCount++;
            if (retryCount <= MAX_RETRIES) {
                log.info("重试失败，失败次数：{}，消息：{}", retryCount, consumeRecord.value());
                // 更新消息头中的重试次数并重新发送到原主题
                sendWithRetryHeader(consumeRecord.topic(), consumeRecord.key(), consumeRecord.value(), retryCount);
            } else {
                log.error("超过最大重试次数，消息：{}", consumeRecord.value());
                // 发送到死信队列
                kafkaTemplate.send(DEAD_LETTER_TOPIC, consumeRecord.key(), consumeRecord.value());
                ack.acknowledge(); // ❗转发到 DLQ 后，手动 ack，防止 Kafka 一直重复投递
            }
        }
    }

    /**
     * Kafka支持事务性消息
     *
     * @param consumeRecord 消费者记录
     * @param ack           手动确认对象
     */
    /*
        Kafka 能做到的是“事务性生产 + 最终一致性消费”，不等于数据库级别的全局分布式事务。如果你要跨 Kafka 与数据库、Redis、HTTP 服务等实现一致性，请考虑：
        ✅ 幂等性设计（业务唯一 key）
        ✅ 手动提交 offset（控制消费确认时机）
        ✅ 本地消息表 + 补偿机制（适合 DDD/Event Sourcing）
        ✅ 事务出错转 DLQ + 监控
     */
    @KafkaListener(topics = "transactional-topic", groupId = "transactional-group")
    public void consumeTransactional(ConsumerRecord<String, String> consumeRecord, Acknowledgment ack) {
        // 开启事务
        kafkaTemplate.executeInTransaction(operations -> {
            try {
                log.info("处理事务消息，主题：{}，key：{}，value: {}",
                        consumeRecord.topic(), consumeRecord.key(), consumeRecord.value());
                // 业务处理
                // .......
                // 发送到其他主题
                operations.send("processed-topic1", consumeRecord.key(), consumeRecord.value()); // ✅ 第一步：发送到其他主题
                operations.send("processed-topic2", consumeRecord.key(), consumeRecord.value()); // ✅ 第一步：发送到其他主题
                operations.send("processed-topic3", consumeRecord.key(), consumeRecord.value()); // ✅ 第一步：发送到其他主题
                // 手动提交偏移量（事务中会自动处理）
                ack.acknowledge();  // ✅ 第二步：提交 offset，表示成功处理

                return true; // 事务成功
            } catch (Exception e) {
                log.error("事务失败，失败原因：{}", e.getMessage());
                // 事务回滚
                throw new RuntimeException(e);
            }
        });
    }

    /**
     * 发送带有重试次数的消息
     *
     * @param topic      目标主题
     * @param key        消息的 key
     * @param value      消息的内容
     * @param retryCount 重试次数
     */
    private void sendWithRetryHeader(String topic, String key, String value, int retryCount) {
        // 创建带有重试次数的消息头
        // 使用 Kafka 的 RecordHeaders 添加重试次数
        RecordHeaders headers = new RecordHeaders();
        headers.add(new RecordHeader("retry-count", String.valueOf(retryCount).getBytes(StandardCharsets.UTF_8)));

        // 构造 Kafka 原生 ProducerRecord
        ProducerRecord<String, String> producerRecord =
                new ProducerRecord<>(topic, null, null, key, value, headers);
        kafkaTemplate.send(producerRecord);
    }

    /**
     * 从消息头中提取重试次数
     *
     * @param headers 消息头
     * @return 重试次数
     */
    public int extractRetryCount(Headers headers) {
        Header retryHeader = headers.lastHeader("retry-count");
        if (retryHeader != null) {
            try {
                // 假设 retry-count 是以字符串形式存的
                String value = new String(retryHeader.value(), StandardCharsets.UTF_8);
                return Integer.parseInt(value);
            } catch (Exception e) {
                // 解析失败默认返回 0
                return 0;
            }
        }
        return 0;
    }
}